
package caiqr.model.fb_asia_odds

//import com.db.five_million.football_match_sporttery_service
//import com.spark.firstApp.SavePrediction
//import com.hdfs.football_match_asia_odds
//import com.spark.firstApp.SavePrediction
import org.apache.spark.rdd.RDD
//import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{SQLContext, DataFrame}
import java.text.SimpleDateFormat
import java.sql.DriverManager
//import java.lang.IllegalArgumentException


// 比较亚盘初盘/终盘盘口+SP相同赛事
object AsiaInitOrCurrOddsSp {


  //// 初盘盘口+赔率相同比赛.
  //1. 公司+赛事
  def asia_init_sp_same_by_company_seasonname(asia_df: DataFrame, mini_cnt: Int) = {
    // 1). 获取所有数据
    val match_list_rdd = asia_df.orderBy(asia_df("match_time").asc).
      selectExpr("company_id", "season_pre", "init_odds","init_home","init_away", "match_time", "match_id","match_result", "asia_init_win_flag")

    val dds_type = "sp_init_sname"
    val new_match_id_list_rdd = compute_initsp_by_company_seasonname(match_list_rdd, dds_type, mini_cnt)
    //new_match_id_list_rdd.collect().foreach(println)

    new_match_id_list_rdd
    // TODO
    // save to DB...
  }




  //// 初盘盘口+赔率相同比赛.
  //2. 公司+所有比赛
  def asia_init_sp_all_same_by_company_seasonname(asia_df: DataFrame, mini_cnt: Int) = {
    // 1). 获取所有数据
    val match_list_rdd = asia_df.orderBy(asia_df("match_time").asc).
      selectExpr("company_id", "init_odds","init_home","init_away", "match_time", "match_id","match_result", "asia_init_win_flag")

    val dds_type = "sp_init_all"
    val new_match_id_list_rdd = compute_initsp_all_by_company_seasonname(match_list_rdd, dds_type, mini_cnt)
    //new_match_id_list_rdd.collect().foreach(println)

    new_match_id_list_rdd
    // TODO
    // save to DB...
  }




  //// 终盘盘口+赔率相同比赛.
  //3. 公司+赛事
  def asia_curr_sp_same_by_company_seasonname(asia_df: DataFrame, mini_cnt: Int) = {
    // 3). 获取所有数据
    val match_list_rdd = asia_df.orderBy(asia_df("match_time").asc).
      selectExpr("company_id", "season_pre", "curr_odds","curr_home","curr_away", "match_time", "match_id","match_result", "asia_curr_win_flag")

    val dds_type = "sp_curr_sname"
    val new_match_id_list_rdd = compute_initsp_by_company_seasonname(match_list_rdd, dds_type, mini_cnt)
    //new_match_id_list_rdd.collect().foreach(println)

    new_match_id_list_rdd
    // TODO
    // save to DB...
  }




  //// 终盘盘口+赔率相同比赛.
  //4. 公司+所有比赛
  def asia_curr_sp_all_same_by_company_seasonname(asia_df: DataFrame, mini_cnt: Int) = {
    // 4). 获取所有数据
    val match_list_rdd = asia_df.orderBy(asia_df("match_time").asc).
      selectExpr("company_id", "curr_odds","curr_home","curr_away", "match_time", "match_id","match_result", "asia_curr_win_flag")

    val dds_type = "sp_curr_all"
    val new_match_id_list_rdd = compute_initsp_all_by_company_seasonname(match_list_rdd, dds_type, mini_cnt)
    //new_match_id_list_rdd.collect().foreach(println)

    new_match_id_list_rdd
    // TODO
    // save to DB...
  }



  //// 初盘盘口+赔率相同, + , 终盘盘口+赔率相同比赛.
  //5. 公司+赛事
  def asia_init_curr_sp_same_by_company_seasonname(asia_df: DataFrame, mini_cnt: Int) = {
    // 4). 获取所有数据
    val match_list_rdd = asia_df.orderBy(asia_df("match_time").asc).
      selectExpr("company_id", "init_odds","init_home","init_away", "curr_odds","curr_home","curr_away", "match_time", "match_id","match_result", "asia_init_win_flag", "asia_curr_win_flag")

    val dds_type = "sp_init_curr_all"
    val new_match_id_list_rdd = compute_all_by_company_seasonname(match_list_rdd, dds_type, mini_cnt)
    //new_match_id_list_rdd.collect().foreach(println)

    new_match_id_list_rdd
    // TODO
    // save to DB...
  }




  //// 初盘盘口+赔率相同, 终盘盘口相同比赛.
  //6. 公司+赛事
  def asia_init_sp_curr_odds_same_by_company_seasonname(asia_df: DataFrame, mini_cnt: Int) = {
    // 1). 获取所有数据
    val match_list_rdd = asia_df.orderBy(asia_df("match_time").asc).
      selectExpr("company_id", "season_pre", "init_odds","init_home","init_away", "curr_odds", "match_time", "match_id","match_result", "asia_init_win_flag","asia_curr_win_flag")

    val dds_type = "sp_init_curr_odds_sname"
    val new_match_id_list_rdd = compute_initsp_curr_odds_by_company_seasonname(match_list_rdd, dds_type, mini_cnt)
    //new_match_id_list_rdd.collect().foreach(println)

    new_match_id_list_rdd
    // TODO
    // save to DB...
  }

  //7. 公司+所有比赛
  def asia_init_sp_curr_odds_all_same_by_company(asia_df: DataFrame, mini_cnt: Int) = {
    // 1). 获取所有数据
    val match_list_rdd = asia_df.orderBy(asia_df("match_time").asc).
      selectExpr("company_id", "init_odds","init_home","init_away", "curr_odds", "match_time", "match_id","match_result", "asia_init_win_flag","asia_curr_win_flag")

    val dds_type = "sp_init_curr_odds_all"
    val new_match_id_list_rdd = compute_initsp_curr_odds_by_company(match_list_rdd, dds_type, mini_cnt)
    //new_match_id_list_rdd.collect().foreach(println)

    new_match_id_list_rdd
    // TODO
    // save to DB...
  }





  def compute_initsp_by_company_seasonname(match_list_rdd:  DataFrame, dds_type: String, mini_cnt: Int): RDD[(String,String)] ={

    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    // 转换为元祖 ("company_id", "season_pre", "init_odds","init_home","init_away", "match_time", "match_id","match_result", "asia_init_win_flag")
    // OUT:
    val tuple_same_init_odds_rdd = match_list_rdd.rdd.map { p =>

      val match_time = p.getString(5)
      val match_time_second = sdf.parse(match_time).getTime

      (s"${p.getString(0)}_${p.getString(1)}_${p.getString(2)}_${p.getString(3)}_${p.getString(4)}",
        (p.getString(6), p.getString(7), p.getString(8), match_time_second.toString()))
    }
    //tuple_same_init_odds_rdd.collect().foreach(println)


    // out:
    val tuple_same_init_odds_index_rdd = tuple_same_init_odds_rdd.zipWithIndex
    //tuple_same_init_odds_index_rdd.collect().foreach(println)


    // out:
    val new_tuple_same_init_odds_index_rdd = tuple_same_init_odds_index_rdd.map(p =>
      //(      p._1._1, ((p._1._2._1, p._2),p._1._2._2)     )
      (p._1._1, (p._2, p._1._2))
    )
    //new_tuple_same_init_odds_index_rdd.collect().foreach(println)


    //OUT:
    //倒叙
    val new_tuple_same_init_odds_index_order_rdd = new_tuple_same_init_odds_index_rdd.groupByKey().map { p =>
      val sortArray = p._2.toArray.sortWith(_._1 > _._1)
      (p._1, sortArray)
    }
    //    new_tuple_same_init_odds_index_order_rdd.collect().foreach{p =>
    //          println(p._1)
    //          p._2.foreach(println)
    //        }
    //new_tuple_same_init_odds_index_order_rdd.collect().foreach(println)


    //OUT:
    val new_tuple_same_init_odds_rdd = new_tuple_same_init_odds_index_order_rdd.map(p =>
      (p._1,
        p._2.map(p => p._2._1.toString()).reduce(_ + "," + _),
        p._2.map(p => p._2._2.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._3.toString()).reduce(_ + "" + _))
    ).filter(p => p._3.length > mini_cnt).map(p =>
//      (dds_type.toString()+"_"+p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()))
      (p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()))
    //new_tuple_same_init_odds_rdd.collect().foreach(println)

    //SaveDds.save_asia_dds_sp_result_to_mysql(new_tuple_same_init_odds_rdd, dds_type)

    new_tuple_same_init_odds_rdd
  }



  def compute_initsp_all_by_company_seasonname(match_list_rdd:  DataFrame, dds_type: String, mini_cnt: Int): RDD[(String,String)] ={

    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    // 转换为元祖 ("company_id", "init_odds","init_home","init_away", "match_time", "match_id","match_result", "asia_init_win_flag")
    // OUT:
    val tuple_same_init_odds_rdd = match_list_rdd.rdd.map { p =>

      val match_time = p.getString(4)
      val match_time_second = sdf.parse(match_time).getTime

      (s"${p.getString(0)}_${p.getString(1)}_${p.getString(2)}_${p.getString(3)}",
        (p.getString(5), p.getString(6), p.getString(7), match_time_second.toString()))
    }
    //tuple_same_init_odds_rdd.collect().foreach(println)


    // out:
    val tuple_same_init_odds_index_rdd = tuple_same_init_odds_rdd.zipWithIndex
    //tuple_same_init_odds_index_rdd.collect().foreach(println)


    // out:
    val new_tuple_same_init_odds_index_rdd = tuple_same_init_odds_index_rdd.map(p =>
      //(      p._1._1, ((p._1._2._1, p._2),p._1._2._2)     )
      (p._1._1, (p._2, p._1._2))
    )
    //new_tuple_same_init_odds_index_rdd.collect().foreach(println)


    //OUT:
    //倒叙
    val new_tuple_same_init_odds_index_order_rdd = new_tuple_same_init_odds_index_rdd.groupByKey().map { p =>
      val sortArray = p._2.toArray.sortWith(_._1 > _._1)
      (p._1, sortArray)
    }
    //    new_tuple_same_init_odds_index_order_rdd.collect().foreach{p =>
    //          println(p._1)
    //          p._2.foreach(println)
    //        }
    //new_tuple_same_init_odds_index_order_rdd.collect().foreach(println)


    //OUT:
    val new_tuple_same_init_odds_rdd = new_tuple_same_init_odds_index_order_rdd.map(p =>
      (p._1,
        p._2.map(p => p._2._1.toString()).reduce(_ + "," + _),
        p._2.map(p => p._2._2.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._3.toString()).reduce(_ + "" + _))
    ).filter(p => p._3.length > mini_cnt).map(p =>
//      (dds_type.toString()+"_"+p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()))
      (p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()))
    //new_tuple_same_init_odds_rdd.collect().foreach(println)

    //SaveDds.save_asia_dds_sp_result_to_mysql(new_tuple_same_init_odds_rdd, dds_type)

    new_tuple_same_init_odds_rdd
  }





  def compute_all_by_company_seasonname(match_list_rdd:  DataFrame, dds_type: String, mini_cnt: Int): RDD[(String,String)] ={

    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    // 转换为元祖 ("company_id", "init_odds","init_home","init_away", "curr_odds","curr_home","curr_away", "match_time", "match_id","match_result", "asia_init_win_flag", "asia_curr_win_flag")
    // OUT:
    val tuple_same_init_odds_rdd = match_list_rdd.rdd.map { p =>

      val match_time = p.getString(7)
      val match_time_second = sdf.parse(match_time).getTime

      (s"${p.getString(0)}_${p.getString(1)}_${p.getString(2)}_${p.getString(3)}_${p.getString(4)}_${p.getString(5)}_${p.getString(6)}",
        (p.getString(8), p.getString(9), p.getString(10), p.getString(11), match_time_second.toString()))
    }
    //tuple_same_init_odds_rdd.collect().foreach(println)


    // out:
    val tuple_same_init_odds_index_rdd = tuple_same_init_odds_rdd.zipWithIndex
    //tuple_same_init_odds_index_rdd.collect().foreach(println)


    // out:
    val new_tuple_same_init_odds_index_rdd = tuple_same_init_odds_index_rdd.map(p =>
      //(      p._1._1, ((p._1._2._1, p._2),p._1._2._2)     )
      (p._1._1, (p._2, p._1._2))
    )
    //new_tuple_same_init_odds_index_rdd.collect().foreach(println)


    //OUT:
    //倒叙
    val new_tuple_same_init_odds_index_order_rdd = new_tuple_same_init_odds_index_rdd.groupByKey().map { p =>
      val sortArray = p._2.toArray.sortWith(_._1 > _._1)
      (p._1, sortArray)
    }
    //    new_tuple_same_init_odds_index_order_rdd.collect().foreach{p =>
    //          println(p._1)
    //          p._2.foreach(println)
    //        }
    //new_tuple_same_init_odds_index_order_rdd.collect().foreach(println)


    //OUT:
    val new_tuple_same_init_odds_rdd = new_tuple_same_init_odds_index_order_rdd.map(p =>
      (p._1,
        p._2.map(p => p._2._1.toString()).reduce(_ + "," + _),
        p._2.map(p => p._2._2.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._3.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._4.toString()).reduce(_ + "" + _))
    ).filter(p => p._3.length > mini_cnt).map(p =>
//      (dds_type.toString()+"_"+p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()+";"+p._5.toString()))
      (p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()+";"+p._5.toString()))
    //new_tuple_same_init_odds_rdd.collect().foreach(println)

    //SaveDds.save_asia_dds_init_curr_sp_result_to_mysql(new_tuple_same_init_odds_rdd, dds_type)

    new_tuple_same_init_odds_rdd
  }







  def compute_initsp_curr_odds_by_company_seasonname(match_list_rdd:  DataFrame, dds_type: String, mini_cnt: Int): RDD[(String,String)] ={

    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    // 转换为元祖 ("company_id", "season_pre", "init_odds","init_home","init_away", "curr_odds", "match_time", "match_id","match_result", "asia_init_win_flag","asia_curr_win_flag")
    // OUT:
    val tuple_same_init_odds_rdd = match_list_rdd.rdd.map { p =>

      val match_time = p.getString(6)
      val match_time_second = sdf.parse(match_time).getTime

      (s"${p.getString(0)}_${p.getString(1)}_${p.getString(2)}_${p.getString(3)}_${p.getString(4)}_${p.getString(5)}",
        (p.getString(7), p.getString(8), p.getString(9),p.getString(10), match_time_second.toString()))
    }
    //tuple_same_init_odds_rdd.collect().foreach(println)


    // out:
    val tuple_same_init_odds_index_rdd = tuple_same_init_odds_rdd.zipWithIndex
    //tuple_same_init_odds_index_rdd.collect().foreach(println)


    // out:
    val new_tuple_same_init_odds_index_rdd = tuple_same_init_odds_index_rdd.map(p =>
      //(      p._1._1, ((p._1._2._1, p._2),p._1._2._2)     )
      (p._1._1, (p._2, p._1._2))
    )
    //new_tuple_same_init_odds_index_rdd.collect().foreach(println)


    //OUT:
    //倒叙
    val new_tuple_same_init_odds_index_order_rdd = new_tuple_same_init_odds_index_rdd.groupByKey().map { p =>
      val sortArray = p._2.toArray.sortWith(_._1 > _._1)
      (p._1, sortArray)
    }
    //    new_tuple_same_init_odds_index_order_rdd.collect().foreach{p =>
    //          println(p._1)
    //          p._2.foreach(println)
    //        }
    //new_tuple_same_init_odds_index_order_rdd.collect().foreach(println)


    //OUT:
    val new_tuple_same_init_odds_rdd = new_tuple_same_init_odds_index_order_rdd.map(p =>
      (p._1,
        p._2.map(p => p._2._1.toString()).reduce(_ + "," + _),
        p._2.map(p => p._2._2.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._3.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._3.toString()).reduce(_ + "" + _))
    ).filter(p => p._3.length > mini_cnt).map(p =>
//      (dds_type.toString()+"_"+p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()+";"+p._5.toString()))
      (p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()+";"+p._5.toString()))
    //new_tuple_same_init_odds_rdd.collect().foreach(println)

    //SaveDds.save_asia_dds_init_curr_sp_result_to_mysql(new_tuple_same_init_odds_rdd, dds_type)

    new_tuple_same_init_odds_rdd
  }





  def compute_initsp_curr_odds_by_company(match_list_rdd:  DataFrame, dds_type: String, mini_cnt: Int): RDD[(String,String)] ={

    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    // 转换为元祖 ("company_id","init_odds","init_home","init_away", "curr_odds", "match_time", "match_id","match_result", "asia_init_win_flag","asia_curr_win_flag")
    // OUT:
    val tuple_same_init_odds_rdd = match_list_rdd.rdd.map { p =>

      val match_time = p.getString(5)
      val match_time_second = sdf.parse(match_time).getTime

      (s"${p.getString(0)}_${p.getString(1)}_${p.getString(2)}_${p.getString(3)}_${p.getString(4)}",
        (p.getString(6), p.getString(7), p.getString(8),p.getString(9), match_time_second.toString()))
    }
    //tuple_same_init_odds_rdd.collect().foreach(println)


    // out:
    val tuple_same_init_odds_index_rdd = tuple_same_init_odds_rdd.zipWithIndex
    //tuple_same_init_odds_index_rdd.collect().foreach(println)


    // out:
    val new_tuple_same_init_odds_index_rdd = tuple_same_init_odds_index_rdd.map(p =>
      //(      p._1._1, ((p._1._2._1, p._2),p._1._2._2)     )
      (p._1._1, (p._2, p._1._2))
    )
    //new_tuple_same_init_odds_index_rdd.collect().foreach(println)


    //OUT:
    //倒叙
    val new_tuple_same_init_odds_index_order_rdd = new_tuple_same_init_odds_index_rdd.groupByKey().map { p =>
      val sortArray = p._2.toArray.sortWith(_._1 > _._1)
      (p._1, sortArray)
    }
    //    new_tuple_same_init_odds_index_order_rdd.collect().foreach{p =>
    //          println(p._1)
    //          p._2.foreach(println)
    //        }
    //new_tuple_same_init_odds_index_order_rdd.collect().foreach(println)


    //OUT:
    val new_tuple_same_init_odds_rdd = new_tuple_same_init_odds_index_order_rdd.map(p =>
      (p._1,
        p._2.map(p => p._2._1.toString()).reduce(_ + "," + _),
        p._2.map(p => p._2._2.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._3.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._3.toString()).reduce(_ + "" + _))
    ).filter(p => p._3.length > mini_cnt).map(p =>
//      (dds_type.toString()+"_"+p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()+";"+p._5.toString()))
      (p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()+";"+p._5.toString()))
    //new_tuple_same_init_odds_rdd.collect().foreach(println)

    //SaveDds.save_asia_dds_init_curr_sp_result_to_mysql(new_tuple_same_init_odds_rdd, dds_type)

    new_tuple_same_init_odds_rdd
  }


}


